package com.adingxiong.cft.controller.kafka;

import com.adingxiong.cft.errors.ApiException;
import io.swagger.annotations.Api;
import org.apache.kafka.clients.producer.RecordMetadata;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.kafka.core.KafkaTemplate;
import org.springframework.transaction.annotation.Transactional;
import org.springframework.web.bind.annotation.GetMapping;
import org.springframework.web.bind.annotation.PathVariable;
import org.springframework.web.bind.annotation.RequestMapping;
import org.springframework.web.bind.annotation.RestController;

/**
 * @ClassName ProducerController
 * @Description TODO
 * @Author xiongchao
 * @Date 2021/3/24 13:27
 **/
@RestController
@Api(tags = "kafka 生产者测试")
@RequestMapping("/kafka")
public class ProducerController {


    @Autowired
    private KafkaTemplate<String,Object> kafkaTemplate;


    @GetMapping("/test/send/{message}")
    public void sendMsg(@PathVariable("message") String message) {
        kafkaTemplate.send("topic1",message);
    }

    @GetMapping("/test/sendcallback/{message}")
    public String  sendMsgCallBack(@PathVariable("message") String message) {
        StringBuilder builder = new StringBuilder();
        kafkaTemplate.send("topic1",message).addCallback( sucess -> {
            RecordMetadata me = sucess.getRecordMetadata();
            builder.append("topic:");
            builder.append(me.topic());
            builder.append("partition:");
            builder.append(me.partition());
            builder.append("offset :" + me.offset());
        }, failuer -> {
            builder.append("消息发送失败:" +failuer.getMessage());
        });
        return builder.toString();
    }


    /*@GetMapping("/test/send1/{message}")
    public void sendMsg1(@PathVariable("message") String message) {
        //通过事物发送消息,失败后面会停止
        kafkaTemplate.executeInTransaction( operations ->  {
            operations.send("topic2",message);
            throw new ApiException("fail");
        });
        //不通过失败发送消息,失败后面消息继续
        kafkaTemplate.send("topic1",message);
        throw new ApiException("fail");
    }*/


    @GetMapping("/send/{input}")
    @Transactional(rollbackFor = RuntimeException.class,transactionManager = "kafkaTransactionManager")
    public String sendToKafka(@PathVariable String input) {
        kafkaTemplate.send("test",input);
        if ("error".equals(input)){
            throw new RuntimeException("this is a error!!!!!!");
        }
        kafkaTemplate.send("test",input+" mmmmmm");
        return "send success!"+input;
    }



}
